Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@docs/CROSS_ENGINE.md`:
- Around line 93-94: The documentation in CROSS_ENGINE.md at the specified lines
overstates the current index handling capabilities. The current implementation
only synthesizes CREATE TABLE IF NOT EXISTS statements and row inserts, but does
not actually recreate indexes. Reword the statement about indexes from claiming
they are created (but not optimized) to accurately reflect that index recreation
is not implemented yet, aligning the documentation with the actual behavior of
the canonical consume path.

In `@internal/app/canonical_consume.go`:
- Around line 19-21: The bufio.Scanner with a 16 MiB buffer limit can cause
token size violations when processing large rows, resulting in mid-stream aborts
and partial replay state. Replace the Scanner-based approach (where sc is
created with NewScanner and Buffer methods are called) with json.Decoder, which
does not impose token size limits on the input stream. This change should be
applied throughout the canonical_consume.go file, including all locations where
the Scanner is currently used for row streaming to ensure consistent behavior
and eliminate the risk of large-row transfer failures.

In `@internal/app/canonical_emit.go`:
- Around line 43-56: The scanned values in the vals slice can contain []byte
which gets marshaled as base64 JSON, but text/JSON/decimal-like values need to
be normalized first. In the loop where the map m is populated from t.Columns,
add normalization logic before assigning each value to m[c.Name] to detect and
transform any []byte values into their appropriate string or typed
representations based on the column type, ensuring that text/JSON/decimal values
are properly converted before being passed to writeJSONL.

In `@internal/app/restore.go`:
- Around line 63-74: The envelope returned from dumps.ReadEnvelope at line 63 is
being ignored (assigned to underscore), which means there is no validation that
the dump's driver matches the current driver before conn.Restore is called with
Clean=true on the first iteration. Capture the envelope instead of ignoring it,
extract the driver information from the envelope/meta, and add a strict driver
match validation before the conn.Restore call at line 74 to ensure the dump is
compatible with the current driver before any destructive operations like clean
are performed.

In `@internal/cli/backup.go`:
- Around line 31-48: The CLI currently accepts the --base flag without enforcing
that it can only be used with the --incremental flag, making the behavior
ambiguous when --base is provided without --incremental. Add a validation check
before or near the existing incremental backup validation that rejects the
combination when baseID is not empty and incremental is false. Return a
user-facing error that explains --base can only be used with --incremental,
similar to the structure of the existing incremental backup error that uses
errs.Error with Op, Code, Cause, and Hint fields.

In `@internal/driver/_mysqlcommon/incremental.go`:
- Around line 101-110: The binlogArgs function does not include SSL/TLS
configuration from the driver.Profile.SSLMode, which means incremental binlog
capture ignores required encryption policies. To fix this, modify the binlogArgs
function signature to accept the binlogBinary parameter (which indicates whether
the binary is mysqlbinlog or mariadb-binlog), create a helper function that maps
p.SSLMode to fork-specific SSL flags (MySQL uses --ssl-mode with
--ssl-ca/--ssl-cert/--ssl-key while MariaDB uses --ssl and
--ssl-verify-server-cert), and append the appropriate SSL flags to the argument
slice before adding the final since.File argument.

In `@internal/driver/postgres/incremental.go`:
- Around line 54-70: The incrementalArgs function contains two issues: the `-D`
argument uses `-` expecting stdout, but pg_receivewal requires an actual
directory path and does not support stdout streaming, so this argument must be
changed to specify a valid directory instead. Additionally, the comment on line
72 incorrectly describes `--no-loop` as controlling WAL segment termination when
it actually controls retry behavior on connection errors, so update the comment
to accurately reflect the flag's true purpose and remove the incorrect
characterization of its function.

In `@internal/jobs/stream_test.go`:
- Around line 177-190: The test TestStream_ConcurrentWriteCloseNoPanic spawns
three goroutines (for Close, Write, and Read operations) inside a loop without
waiting for them to complete, allowing the test to return before the concurrent
operations are actually executed and checked for race conditions. Add a
sync.WaitGroup to synchronize the spawned goroutines: initialize it before the
loop, call Add(3) for each iteration to account for the three goroutines, invoke
Done() at the end of each spawned goroutine function, and call Wait() after the
loop to ensure all goroutines complete before the test returns.

In `@internal/jobs/stream.go`:
- Around line 33-35: The Write method currently enqueues the entire input byte
slice p as a single chunk, which allows large writes to consume unbounded memory
and bypass the capChunks backpressure mechanism. Modify the Write method to
split the incoming p into appropriately sized chunks (respecting the intended
chunk size limit of ~1MB as mentioned in the NewStream documentation) and
enqueue each chunk individually. This ensures that large Write calls are
properly throttled and the buffer backpressure works as designed, rather than
allowing any single Write to circumvent the buffer bounds.
- Around line 72-73: The Read method of the Stream struct must handle
zero-length reads explicitly to comply with the io.Reader contract. Add a check
at the beginning of the Read method (before the overflow check) to return (0,
nil) immediately when len(p) == 0, ensuring that callers probing with empty
buffers do not block on the select statement that waits for data from s.ch or
s.done.

---

Outside diff comments:
In `@internal/app/sync.go`:
- Around line 97-104: The code on lines 97-104 has a logic issue where closing
the stream to unblock the backup goroutine can cause it to return
io.ErrClosedPipe, which then masks the actual restore error. Fix this by
checking if restoreErr is not nil and returning it first (since that's the real
operation error), and only return backupErr if restoreErr is nil. Alternatively,
handle the specific case where backupErr is io.ErrClosedPipe (the expected error
from intentionally closing the stream) and suppress it, allowing the real
restoreErr to be returned instead.

---

Nitpick comments:
In `@internal/app/cdc.go`:
- Around line 88-93: The jobID is hardcoded to the constant string "cdc" on line
88, causing all source/target sync pairs to share the same state file and
overwrite each other's resume positions. Replace the hardcoded jobID with a
dynamic value that incorporates both opt.From and opt.To (the source and target)
to create a unique key for each distinct sync pair. This ensures each
source/target combination maintains its own separate state file for proper
resume functionality.
- Around line 91-93: The error handling in the loadCDCState call currently
ignores all errors when err is nil, which masks real failures like malformed or
unreadable state files. Modify the error handling logic in the loadCDCState
block to specifically check if the error is a missing-state error (typically a
file-not-found type error) and only ignore that specific error type. For all
other errors returned by loadCDCState, ensure they are surfaced by logging or
returning them rather than silently continuing with the default state. This way,
real state file corruption or access issues will be caught instead of silently
resetting progress.

In `@internal/driver/postgres/incremental_test.go`:
- Around line 14-51: Add integration test coverage for the BackupIncremental
method to catch pg_receivewal command invocation regressions. After the existing
slot creation and base backup operations in TestIncremental_SlotAndLSNCapture
(or in a new complementary test), invoke the BackupIncremental method on the
*Conn object to exercise the WAL streaming via pg_receivewal command, capturing
its output and verifying successful execution. This ensures that CLI contract
changes in the pg_receivewal command arguments and environment setup will be
detected during test execution.